package io.shdxhl.simplerpc.client.request;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.shdxhl.simplerpc.client.provider.ServiceProvider;
import io.shdxhl.simplerpc.code.cache.SimpleProviderCache;
import io.shdxhl.simplerpc.code.netty.ChannelMapping;
import io.shdxhl.simplerpc.code.netty.RpcRequestPromise;
import io.shdxhl.simplerpc.code.netty.codec.*;
import io.shdxhl.simplerpc.code.netty.handler.RpcResponseHandler;
import io.shdxhl.simplerpc.code.netty.request.RpcRequest;
import io.shdxhl.simplerpc.code.netty.response.RpcResponse;
import io.shdxhl.simplerpc.code.netty.strategy.LoadBalanceStrategy;
import io.shdxhl.simplerpc.code.netty.strategy.StrategyProvider;
import io.shdxhl.simplerpc.code.utils.RequestUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * 发送请求，返回当前业务结果
 */
@Slf4j
@Component
public class SimpleRequestManager {

    @Autowired
    private SimpleProviderCache simpleProviderCache;

    @Autowired
    private StrategyProvider strategyProvider;

    public RpcResponse sendRequest(RpcRequest request) {
        String className = request.getClassName();
        String methodName = request.getMethodName();
        List<ServiceProvider> providers = simpleProviderCache.get(className);
        if(providers.isEmpty() || providers.size() == 0){
            log.info("接口{}没有提供者",request.getClassName());
            throw  new RuntimeException("接口：" + request.getClassName() + " 没有提供者");
        }

        //负载均衡
        LoadBalanceStrategy strategy = strategyProvider.getStrategy();
        ServiceProvider serviceProvider = strategy.select(providers);
        return requestByNetty(request, serviceProvider);
    }

    private RpcResponse requestByNetty(RpcRequest request, ServiceProvider serviceProvider) {
        Channel channel = null;
        if(!RequestUtil.hasChannel(serviceProvider.getServiceIp(), serviceProvider.getRpcPort())){
            EventLoopGroup group = new NioEventLoopGroup();
            Bootstrap bootstrap = new io.netty.bootstrap.Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            //add encoder
                            pipeline.addLast("frameEncoder", new FrameEncoder());
                            pipeline.addLast("requestEncoder", new RpcRequestEncoder());

                            //add decoder
                            pipeline.addLast("frameDecoder", new FrameDecoder());
                            pipeline.addLast("responseDecoder", new RpcResponseDecoder());

                            //add handler
                            pipeline.addLast("responseHandler", new RpcResponseHandler());
                        }
                    });
            try {
                ChannelFuture channelFuture = bootstrap.connect(serviceProvider.getServiceIp(), serviceProvider.getRpcPort()).sync();
                channel = channelFuture.channel();
                //添加本地内存中
                ChannelMapping channelMapping = new ChannelMapping(serviceProvider.getServiceIp(), serviceProvider.getRpcPort(), channel);
                RequestUtil.addChannel(channelMapping.getIpWithPort(), channelMapping);
            } catch (InterruptedException e) {
                log.error("can't connect to {}:{}",serviceProvider.getServiceIp(),serviceProvider.getRpcPort());
                throw new RuntimeException("con't connect to " + serviceProvider.getServiceIp() + ":" + serviceProvider.getRpcPort());
            }
        }

        //建立连接
        try {
            //从内存中找到当前Channel
            channel = RequestUtil.getChannel(serviceProvider.getServiceIp(), serviceProvider.getRpcPort());
            if(channel != null && channel.isActive()){
                RpcRequestPromise requestPromise = new RpcRequestPromise(channel.eventLoop());
                RequestUtil.addRequest(request.getRequestId(), requestPromise);
                channel.writeAndFlush(request);

                log.info("SimpleRequestManager 获取异步的结果");
                //获取结果，如何异步获取结果等待回调
                RpcResponse rpcResponse =(RpcResponse) requestPromise.get();
                RequestUtil.removeRequest(rpcResponse.getRequestId());
                return rpcResponse;
            }
        } catch (Exception e) {
            log.error("requestByNetty rpc call error:", e);
            RequestUtil.removeRequest(request.getRequestId());
            throw new RuntimeException(e);
        }
        return null;
    }
}
